package com.tn.rtdata.jms;

import java.util.List;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.TextMessage;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.IllegalStateException;

import com.tn.openrpc.RemoteMethod;
import com.tn.openrpc.serialization.SerializationException;
import com.tn.openrpc.serialization.Serializer;
import com.tn.rtdata.EventDrivenDataSource;
import com.tn.rtdata.FillException;
import com.tn.rtdata.Session;

/**
 * An implementation of <code>DataSource</code> that uses JMS to receive <code>FillEvent</code>s.
 */
public class JmsDataSource<T> extends EventDrivenDataSource<T>
{
  private static final Logger LOGGER = LoggerFactory.getLogger(JmsDataSource.class);

  private static final long DEFAULT_TIMEOUT = 1000;
  private static final long SHUTDOWN_SLEEP_INTERVAL = 100;

  private static int threadCount = 0;

  private ConnectionFactory connectionFactory;
  private String fillEventDestination;
  private FillEventMonitor fillEventMonitor;
  private Serializer<String> serializer;
  private long timeout;

  /**
   * Creates a new <code>JmsDataSource</code>.
   *
   * @param session                    the associated session.
   * @param fillName                   the name of the fill.
   * @param fillParameters             the parameters of the fill.
   * @param fillStartMethod            the <code>RemoteMethod</code> that starts the fill.
   * @param fillStopMethod             the <code>RemoteMethod</code> that stops the fill.
   * @param dataEventStatsUpdateMethod the <code>RemoteMethod</code> that called to update the data-event stats.
   * @param connectionFactory          the connection factory used to receive fill event messages.
   * @param fillEventDestination       the destination used to receive fill event messages.
   * @param serializer                 the serializer used to deserialize the fill events.
   */
  public JmsDataSource(
    Session session,
    String fillName,
    List<?> fillParameters,
    RemoteMethod fillStartMethod,
    RemoteMethod fillStopMethod,
    RemoteMethod dataEventStatsUpdateMethod,
    ConnectionFactory connectionFactory,
    Destination fillEventDestination,
    Serializer<String> serializer
  )
  {
    this(
      session,
      fillName,
      fillParameters,
      fillStartMethod,
      fillStopMethod,
      dataEventStatsUpdateMethod,
      connectionFactory,
      fillEventDestination.toString(),
      serializer
    );
  }

  /**
   * Creates a new <code>JmsDataSource</code>.
   *
   * @param session                    the associated session.
   * @param fillName                   the name of the fill.
   * @param fillParameters             the parameters of the fill.
   * @param fillStartMethod            the <code>RemoteMethod</code> that starts the fill.
   * @param fillStopMethod             the <code>RemoteMethod</code> that stops the fill.
   * @param dataEventStatsUpdateMethod the <code>RemoteMethod</code> that called to update the data-event stats.
   * @param connectionFactory          the connection factory used to receive fill event messages.
   * @param fillEventDestination       the destination used to receive fill event messages.
   * @param serializer                 the serializer used to deserialize the fill events.
   */
  public JmsDataSource(
    Session session,
    String fillName,
    List<?> fillParameters,
    RemoteMethod fillStartMethod,
    RemoteMethod fillStopMethod,
    RemoteMethod dataEventStatsUpdateMethod,
    ConnectionFactory connectionFactory,
    String fillEventDestination,
    Serializer<String> serializer
  )
  {
    super(session, fillName, fillParameters, fillStartMethod, fillStopMethod, dataEventStatsUpdateMethod, fillEventDestination);

    this.connectionFactory = connectionFactory;
    this.fillEventDestination = fillEventDestination;
    this.serializer = serializer;
    this.fillEventMonitor = new FillEventMonitor();
    this.timeout = DEFAULT_TIMEOUT;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void startFillEventMonitor()
  {
    Thread fillEventThread = new Thread(fillEventMonitor, "fill-event-monitor-" + threadCount++);
    fillEventThread.setDaemon(true);
    fillEventThread.start();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void stopFillEventMonitor() throws FillException
  {
    fillEventMonitor.stop();
  }

  /**
   * An implementation of <code>Runnable</code> that processes fill event messages.
   */
  private class FillEventMonitor implements Runnable
  {
    private boolean running = false;
    private boolean stopped = true;

    /**
     * {@inheritDoc}
     */
    @Override
    public void run()
    {
      JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
      jmsTemplate.setReceiveTimeout(timeout);

      running = true;
      stopped = false;

      while (running)
      {
        try
        {
          Message message = jmsTemplate.receive(fillEventDestination);

          if (message == null)
          {
            continue;
          }

          if (!(message instanceof TextMessage))
          {
            LOGGER.error("Unhandled message type encountered: {}", message);
            continue;
          }

          LOGGER.debug("Message received: {}", message);

          Object messageContent = serializer.deserialize(((TextMessage)message).getText());
          if (messageContent instanceof Iterable)
          {
            for (Object messageContentItem : (Iterable)messageContent)
            {
              processMessageContent(messageContentItem);
            }
          }
          else
          {
            processMessageContent(messageContent);
          }
        }
        catch (SerializationException e)
        {
          LOGGER.error("An error occurred de-serializing the message contents.", e);
        }
        catch (JMSException e)
        {
          LOGGER.error("An error occurred reading the message contents.", e);
        }
        catch (IllegalStateException e)
        {
          //Ignore - this occurs during shutdown if the
        }
      }

      stopped = true;
      threadCount--;
    }

    /**
     * Stops this <code>FillEventMonitor</code>.
     */
    public void stop()
    {
      running = false;

      //Block while the fill event monitor thread is running.
      while (!stopped)
      {
        try
        {
          Thread.sleep(SHUTDOWN_SLEEP_INTERVAL);
        }
        catch (InterruptedException e)
        {
          //Ignore.
        }
      }
    }
  }
}